Flask SQLAlchemy で REST API を作ってみた

ORMとは

ORM（Object-Relational Mapping）とは、データベースとオブジェクト指向プログラミング言語とのマッピングを行うことを指します。
ただこれだけ聞くと、イメージしづらいですが、簡単に言うと、SQL文を直接記述せずに、通常のオブジェクトを扱うようにデータベースを扱うことができるということです。
Pythonでは、ORMモジュールの1つに SQLAlchemy があります。

SQLAlchemy

SQLAlchemyは、Pythonでポピュラーに使用されているORMライブラリの1つのようです。
SQLite、Postgresql、MySQL、Oracleなどさまざまなエンジンにも対応しており、PythonでORM、DBの処理を行うならSQLAlchemyを第一候補で考えれば良さそうです。
今回は、Flaskを用いてAPIとしてDBの情報を扱いたいので、FlaskにSQLAlchemyの機能を追加したFlask SQLAlchemyを使用します。

インストール

FlaskでREST APIを作成するので、Flask-RESTfulと取得した情報を扱いやすいようにflask-marshmallow もインストールします。

$ pip install Flask, Flask-RESTful, Flask-SQLAlchemy, flask-marshmallow 

DBの準備

Dockerでmaridbを立ち上げます。 下記の内容でファイルを準備し、起動させます。
docker-compose.yml

version: "3"
services:
  db:
    image: mariadb
    restart: always
    ports:
      - 3306:3306
    command: --port 3306
    environment:
      - MYSQL_ROOT_PASSWORD=hogehoge.morita
      - MYSQL_DATABASE=flask_test

mariadbの起動

$ docker-compose up -d

Flask側でのDB設定

インスタンスを生成し、初期化を行う関数を定義します。
この関数は、app.pyから実行されます。
database.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()


def init_db(app):
    db.init_app(app)
    

続いて、初期データを予め投入したいので、関数を準備します。
lib/create_init.py

from models.user  import User
from database import db

def create_init(app):
    with app.app_context():
        db.drop_all()
        db.create_all()
        user_create()
        db.session.commit()
    
def user_create():
    # テストデータ
    from test.data.user import users
    for i in users:
        u  = User(name=i["name"])
        db.session.add(u)

テストデータは、test/data/user.pyの中に記述します。
test/data/user.py

users = [
    { "name" : "山田太郎"},
    { "name" : "上田二郎"},
    { "name" : "田中三郎"}
]

DBの接続情報はconfig.pyで定義します。
接続情報は環境変数から読み出すようにしていますので.envの中に記述します。
config.py

class DevConfig:
    # SQLAlchemy
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://{user}:{password}@{host}/{dbname}?charset=utf8'.format(**{
        'user': os.getenv('DB_USER', 'root'),
        'password': os.getenv('DB_PASSWORD', ''),
        'host': os.getenv('DB_HOST', 'localhost'),
        'dbname': os.getenv('DB_NAME')
    })
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_ECHO = False

Config = DevConfig

.env

DB_USER="root"
DB_PASSWORD="hogehoge.morita"
DB_HOST="localhost"
DB_NAME="flask_test"

ORMとして扱うモデルを定義

続いて、ORMとして扱うモデルを定義します。先程、定義したdbが必要なので、database.pyからimportします。
また、Marshmallowを用いることで、出力フォーマットが簡単に指定ができますので、こちらも定義します。
REST API用のクラスUserapiもここで定義します。
models/user.py

from flask import jsonify, abort, request
from flask_restful import Resource
from datetime import datetime
from database import db
from app import ma

class User(db.Model):

    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    device_id = db.Column(db.String(255), nullable=False, default="hoge")
    name = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at = db.Column(db.DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)

class UserSchema(ma.Schema):
    class Meta:
        # Fields to expose
        fields = ("id", "name", "created_at")
    

users_schema = UserSchema(many=True)


class Userapi(Resource):
    def get(self):
        """
        ユーザを１件取得
        """
        id = request.args.get('id')
        result = User.query.filter_by(id=id).all()
        if len(result) != 0: 
            return jsonify({"users" : users_schema.dump(result)})
        else:
            abort(404)

    def post(self):
        """
        ユーザ登録
        """
        #ユーザを追加
        print(request.json)
        users = request.json["users"]
        for i in users:
            u  = User(name=i["name"])
            db.session.add(u)
        db.session.commit()

        return '', 204

    def put(self):
        """
        ユーザ更新
        """
        id = request.json["id"]
        user = User.query.filter_by(id=id).first()
        
        if user : 
            for key, val in request.json.items():
                setattr(user, key, val)
            db.session.commit()
        else:
            abort(404)

        return '', 204

    def delete(self):
        """
        ユーザ削除
        """
        id = request.args.get('id')
        user = User.query.filter_by(id=id).first()
        if user :
            db.session.delete(user)
            db.session.commit()
            return '', 204
        else:
            abort(404)

app.pyを作成

では、定義したDBの初期化関数、データの投入する関数を実行し、Flaskでサーバを起動させます。
app.py

# coding: utf-8
from flask import Flask
from flask_restful import Api
from flask_marshmallow import Marshmallow

from database import init_db
from lib.create_init import create_init
from models.user import Userapi


app = Flask(__name__)
app.config.from_object('config.Config')

# DB init
init_db(app)
api = Api(app)
ma = Marshmallow(app)
create_init(app)

# API add
api.add_resource(Userapi, '/user')


if __name__ == "__main__":
    app.run(debug=True)

$ flask run

動作確認

正しく動作しているかを確認します。 GETメソッドでユーザ情報1件取得してみます。
user_test.py

import requests

URL = "http://127.0.0.1:5000/"

# GET test
id=1
res = requests.get(URL+"user?id={}".format(id))
print(res.json())

すると、確かにidが1のユーザ情報が取得できていることが確認できます。

{'users': [{'created_at': '2021-10-21T09:45:32', 'id': 1, 'name': '山田太郎'}]}

最後に

Flask-SQLAlchemyを用いて簡易的ですが、ORMを行ってみました。SQL文を記述しなくても良いのはとても便利ですね。
一方で、内部処理がブラックボックスとなっているので、SQLのチューニングする際などにはあまり向いていないようです。その際は従来通りSQL文で記述したほうが良さそうです。
ただ、個人使用であれば、オブジェクトとしてDBを扱えるのはとても便利ですので、今後も使用していきたいと思います。

作成したコード